package function;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;


/**
 * SELECT userId, COUNT(*)
 * FROM clicks
 * GROUP BY SESSION(clicktime, INTERVAL '30' MINUTE), userId
 *
 * select userId, count(*) from clicks group by session(clicktime,interval '30' minute) ,userId
 *
 */

/**
 *  * @param <K> Type of the key.
 *  * @param <I> Type of the input elements.
 *  * @param <O> Type of the output elements.
 */
public class StartEndDuration extends KeyedProcessFunction<String, Tuple2<String,String>,Tuple2<String,Long>> {
    private ValueState<Long> startTime;


    @Override
    public void open(Configuration parameters) throws Exception {
        startTime =getRuntimeContext().getState(new ValueStateDescriptor<Long>("startTime",Long.class));
    }

    @Override
    public void processElement(Tuple2<String, String> in, Context ctx, Collector<Tuple2<String, Long>> out) throws Exception {

        switch (in.f1) {
            case "START":
                startTime.update(ctx.timestamp());
                ctx.timerService().registerEventTimeTimer(ctx.timestamp()+4*60*60*1000);
                break;
            case "END":
                Long sTime =startTime.value();
                if (sTime!=null){
                    out.collect(Tuple2.of(in.f0,ctx.timestamp()-sTime));
                    startTime.clear();
                }
            default:
        }
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<String, Long>> out) throws Exception {
        startTime.clear();
    }
}
